package com.lb.bigdata.spark.core.p2

import java.sql.{Driver, DriverManager}

import com.lb.bigdata.spark.core.p2._02CombineByKeyOps.Student
import com.lb.bigdata.util.db.ConnectionPool
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object _05ForeachPartitionOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
                .setMaster("local[*]")
                .setAppName(s"${_05ForeachPartitionOps.getClass.getSimpleName}")
        val sc = new SparkContext(conf)
        val stu = sc.parallelize(List(
            Student("郭雪磊", 18, "山东"),
            Student("单 松", 20, "山东"),
            Student("刘宇航", 18, "河北"),
            Student("王健", 18, "河南"),
            Student("许迎港", 18, "河北"),
            Student("元永劫", 18, "黑龙江"),
            Student("林博", 18, "黑龙江"),
            Student("李佳奥", 18, "河南"),
            Student("冯世明", 18, "黑龙江"),
            Student("肖楚轩", 18, "山东"),
            Student("张皓", 18, "河南"),
            Student("冯岩", 18, "黑龙江")
        ), 2)

        val pro2stu = stu.map(stu => (stu.province, 1)).reduceByKey(_+_)
        //rdd2db1(pro2stu)
        //rdd2db2(pro2stu)
        rdd2db3(pro2stu)
    }

    def rdd2db1(pro2stu:RDD[(String, Int)]): Unit ={
        pro2stu.foreach{case (province, num) => {
            //1.加载驱动
            classOf[Driver]
            //2.创建连接
            val conn = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306/test",
                "root",
                "root"
            )
            //3.获得执行器
            val sql = "insert into province(province, num) values (?, ?)"
            val ps = conn.prepareStatement(sql)
            ps.setString(1, province)
            ps.setInt(2, num)
            //4.执行sql
            ps.execute()
            //5.封装结果集
            //6.释放资源
            ps.close()
            conn.close()
        }}
    }

    //分区加批量操作
    def rdd2db2(pro2stu:RDD[(String, Int)]): Unit ={
        pro2stu.foreachPartition(partition => {
            classOf[Driver]
            val conn = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306/test",
                "root",
            "root"
            )
            val sql = "insert into province(province, num) values (?, ?)"
            val ps = conn.prepareStatement(sql)

            partition.foreach{case (province, num) => {
                ps.setString(1, province)
                ps.setInt(2, num)
                ps.addBatch()
            }}
            ps.executeBatch()
            ps.close()
            conn.close()
        })
    }

    //调用线程池
    def rdd2db3(pro2stu:RDD[(String, Int)]): Unit ={
        pro2stu.foreachPartition(partition => {
            val conn = ConnectionPool.getConnection()
            val sql = "insert into province(province, num) values (?, ?)"
            val ps = conn.prepareStatement(sql)

            partition.foreach{case (province, num) => {
                ps.setString(1, province)
                ps.setInt(2, num)
                ps.addBatch()
            }}
            ps.executeBatch()
            ps.close()
            conn.close()
        })
    }
}
